package com.shujia.flink.source

import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.common.RuntimeExecutionMode
object Demo1ListSource {
  def main(args: Array[String]): Unit = {

    //创建flink环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
     * flink 执行模式
     * RuntimeExecutionMode.BATCH： 批处理模式，只能用于有界流，计算输出最终的结果
     * RuntimeExecutionMode.STREAMING  : 流处理模式，可以用于有界流也可以用于五界流，输出连续的结果
     */
    env.setRuntimeMode(RuntimeExecutionMode.BATCH)

    /**
     * 集合本地集合source   -  有界流
     *
     * 当读取的数据源是一个有界流时，flink处理完数据就结束了
     *
     */

    val linesDS: DataStream[String] = env.fromCollection(List("java,spark", "java,hadoop", "java"))

    linesDS
      .flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)
      .sum(1)
      .print()


    env.execute()



  }

}
